package com.blink.curator.recipe.listener;

import com.blink.base.BaseTestSample;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.CreateMode;
import org.junit.jupiter.api.Test;

import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * <Description> 子节点监听</Description>
 * <ClassName> PathChildrenCacheSample</ClassName>
 *
 * @Author liuxianzhao
 * @Date 2018年01月13日 16:37
 */
public class PathChildrenCacheSample extends BaseTestSample {
    @Test
    public void testPathChildrenCache() throws Exception {
        CuratorFramework client = getZkCurator();
        client.start();
        String path = "/zk-book/testPathChildrenCache";
        PathChildrenCache cache = new PathChildrenCache(client, path, true);

        /**
         * NORMAL: 初始时为空。
         BUILD_INITIAL_CACHE: 在这个方法返回之前调用rebuild()。
         POST_INITIALIZED_EVENT: 当Cache初始化数据后发送一个PathChildrenCacheEvent.Type#INITIALIZED事件
         */
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client,
                                   PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case CHILD_ADDED:
                        logger.info("CHILD_ADDED," + event.getData().getPath());
                        break;
                    case CHILD_UPDATED:
                        logger.info("CHILD_UPDATED," + event.getData().getPath());
                        break;
                    case CHILD_REMOVED:
                        logger.info("CHILD_REMOVED," + event.getData().getPath());
                        break;
                    default:
                        break;
                }
            }
        });
        client.create().withMode(CreateMode.PERSISTENT).forPath(path);
        sleep(1);
        client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
        sleep(1);
        client.setData().forPath(path + "/c1");
        sleep(1);
        client.delete().forPath(path + "/c1");
        sleep(1);
        client.delete().forPath(path);
        sleep();
    }

    @Test
    public void testPathChildrenCacheNoCacheData() throws Exception {
        CuratorFramework client = getZkCurator();
        client.start();
        String path = "/zk-book/testPathChildrenCacheNoCacheData";
        PathChildrenCache cache = new PathChildrenCache(client, path, false);
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client,
                                   PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case CHILD_ADDED:
                        logger.info("CHILD_ADDED," + event.getData().getData());
                        break;
                    case CHILD_UPDATED:
                        logger.info("CHILD_UPDATED," + event.getData().getData());
                        break;
                    case CHILD_REMOVED:
                        logger.info("CHILD_REMOVED," + event.getData().getPath());
                        break;
                    default:
                        break;
                }
            }
        });
        client.create().withMode(CreateMode.PERSISTENT).forPath(path);
        sleep(1);
        client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
        sleep(1);
        client.setData().forPath(path + "/c1");
        sleep(1);
        client.delete().forPath(path + "/c1");
        sleep(1);
        client.delete().forPath(path);
        sleep();
    }

    @Test
    public void testPathChildrenCacheExecutor() throws Exception {
        CuratorFramework client = getZkCurator();
        client.start();
        String path = "/zk-book/testPathChildrenCacheExecutor";
        ExecutorService tp = new ThreadPoolExecutor(2, 2,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
        ;
        logger.info(Thread.currentThread().getName());
        PathChildrenCache cache = new PathChildrenCache(client, path, true, false, tp);
        cache.start(PathChildrenCache.StartMode.NORMAL);
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client,
                                   PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case CHILD_ADDED:
                        logger.info("CHILD_ADDED," + event.getData().getPath());
                        logger.info("tname: " + Thread.currentThread().getName());
                        break;
                    case CHILD_UPDATED:
                        logger.info("CHILD_UPDATED," + event.getData().getPath());
                        break;
                    case CHILD_REMOVED:
                        logger.info("CHILD_REMOVED," + event.getData().getPath());
                        break;
                    default:
                        break;
                }
            }
        });
        sleep(1);
        client.create().withMode(CreateMode.PERSISTENT).forPath(path);
        sleep(1);
        client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
        sleep(1);
        client.delete().forPath(path + "/c1");
        sleep(1);
        client.delete().forPath(path);
        sleep();
    }

    @Test
    public void testPathChildrenCacheNormal() throws Exception {
        CuratorFramework client = getZkCurator();
        client.start();
        String path = "/zk-book/testPathChildrenCacheNormal";
        PathChildrenCache cache = new PathChildrenCache(client, path, true);
        cache.start(PathChildrenCache.StartMode.NORMAL);
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client,
                                   PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case INITIALIZED:
                        logger.info("INITIALIZED,{}", Optional.ofNullable(event.getData()).orElse(null));
                        break;
                    default:
                        break;
                }
            }
        });
        sleep();
    }

    @Test
    public void testPathChildrenCacheBuildInitialCache() throws Exception {
        CuratorFramework client = getZkCurator();
        client.start();
        String path = "/zk-book/testPathChildrenCacheBuildInitialCache";
        client.create().withMode(CreateMode.PERSISTENT).forPath(path);
        client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
        PathChildrenCache cache = new PathChildrenCache(client, path, true);
        cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client,
                                   PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case INITIALIZED:
                        logger.info("INITIALIZED,{}", Optional.ofNullable(event.getData()).orElse(null));
                        break;
                    case CHILD_ADDED:
                        logger.info("CHILD_ADDED," + event.getData().getPath());
                        logger.info("tname: " + Thread.currentThread().getName());
                        break;
                    case CHILD_UPDATED:
                        logger.info("CHILD_UPDATED," + event.getData().getPath());
                        break;
                    case CHILD_REMOVED:
                        logger.info("CHILD_REMOVED," + event.getData().getPath());
                        break;
                    default:
                        break;
                }
            }
        });
        client.setData().forPath(path + "/c1");
        sleep(1);
        client.delete().forPath(path + "/c1");
        sleep(1);
        client.delete().forPath(path);
        sleep();
    }

}
